package com.atguigu.flink.exec1;

import com.atguigu.flink.pojo.AdsClickLog;
import com.atguigu.flink.pojo.UserBehavior;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashSet;
import java.util.Set;

/**
 * Created by Smexy on 2022/12/14
 *
 */
public class Demo4_AdsClick
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setRuntimeMode(RuntimeExecutionMode.BATCH);
        
                env
                   .readTextFile("data/AdClickLog.csv")
                   .map(new MapFunction<String, AdsClickLog>()
                   {
                       @Override
                       public AdsClickLog map(String value) throws Exception {
                           String[] words = value.split(",");
                            return  new AdsClickLog(
                               Long.valueOf(words[0]),
                               Long.valueOf(words[1]),
                               words[2],
                               words[3],
                               Long.valueOf(words[4])
                           );

                       }
                   })
                   .map(adsClick -> Tuple2.of(adsClick.getProvince() + "_" + adsClick.getAdsId(),1)).returns(new TypeHint<Tuple2<String, Integer>>() {})
                   .keyBy(t -> t.f0)
                   .sum(1)
                   .print();



        
                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }
}
